package com.baishancloud.log.murloc

import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration

import java.util.Objects
import scala.collection.mutable
import scala.util.matching.Regex

/**
 * 根据指定条件过滤数据
 *
 * @author ziqiang.wang 
 * @date 2021/12/21 19:17
 */
class SomeFilter(parameterTool: ParameterTool) extends RichFilterFunction[Origin] {

  //从url中提取码率
  val pattern: Regex = """.*_(\d+)[._]mp4.*""".r("rate")
  //输入参数，域名
  val domainSet: mutable.Set[String] = mutable.Set[String]()
  val contentTypeSet: Set[String] = Set[String]("video/mp4", "video/mp2t")

  override def open(parameters: Configuration): Unit = {
    parameterTool.get(domainIds).split(";").foreach((x: String) => domainSet.add(x.split(":")(0)))
  }

  override def filter(value: Origin): Boolean = {
    val b1: Boolean = domainSet.contains(value.domain())
    val b2: Boolean = contentTypeSet.contains(value.contentType())
    val b3: Boolean = !value.ip().startsWith("127")
    val b4: Boolean = Objects.equals(value.httpMethod(), "GET")
    val b5: Boolean = value.httpCode() >= 200 && value.httpCode() <= 300
    val b6: Boolean = value.responseTime() > 0
    val b7: Boolean = value.bodyBytesSent() > 0
    val b8: Boolean = Objects.equals(value.xPeer(), "-")
    val b9: Boolean = pattern.findFirstMatchIn(value.url()).map(_.group("rate").toInt).fold(false)(_ > 0)
    b1 && b2 && b3 && b4 && b5 && b6 && b7 && b8 && b9
  }
}
